package com.xiguaji.tiktok.task.keyword.service;

import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.gugee.tiktok.common.model.dal.blogger.AwemeCommentRepository;
import com.gugee.tiktok.common.model.dal.blogger.AwemeRepository;
import com.gugee.tiktok.common.model.dal.blogger.BloggerLogRepository;
import com.gugee.tiktok.common.model.dal.blogger.BloggerMainRepository;
import com.gugee.tiktok.common.model.dto.SimpleAweme;
import com.gugee.tiktok.common.model.dto.SlimBloggerDto;
import com.gugee.tiktok.common.model.mysql.AwemeComment;
import com.gugee.tiktok.common.model.mysql.BloggerLog;
import com.gugee.tiktok.data.common.msg.KeywordMessage;
import com.xiguaji.boot.toolkit.DateTimeUtils;
import com.xiguaji.boot.toolkit.JsonUtil;
import com.xiguaji.tiktok.task.common.config.AppConfig;
import com.xiguaji.tiktok.task.common.job.TaskParam;
import com.xiguaji.tiktok.task.common.producer.AbstractProducerHandler;
import com.xxl.job.core.log.XxlJobLogger;
import lombok.extern.slf4j.Slf4j;
import lombok.val;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;

import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;

/**
 * Created with IntelliJ IDEA.
 * Description:
 * User: luolifeng
 * Date: 2020-10-28 15:16
 */
@Service
@Slf4j
public class KeywordService extends AbstractProducerHandler<KeywordMessage> {


    @Autowired
    @Qualifier("kafkaTemplateKeyword")
    private KafkaTemplate<Long, KeywordMessage> kafkaTemplateKeyword;
    @Autowired
    BloggerMainRepository bloggerMainRepository;
    @Autowired
    AppConfig appConfig;
    @Autowired
    AwemeRepository awemeRepository;
    @Autowired
    AwemeCommentRepository awemeCommentRepository;
    @Autowired
    BloggerLogRepository bloggerLogRepository;
    @Override
    public List<KeywordMessage> findMessageObject(BloggerLog bloggerLog, TaskParam taskParam) {
        //查询播主排行中的视频，
//        List<SlimBloggerDto> bloggers = bloggerMainRepository.findSlimBloggersForAuditStatusAndCount(((int)bloggerLog.getCurrentBloggerId()-1)*taskParam.getRecords(),taskParam.getRecords());
//
//        long condition = DateTimeUtils.second(LocalDateTime.now().minusDays(appConfig.getLatestDays()));
        List<KeywordMessage> rst=new ArrayList<>(taskParam.getRecords()*50);
//        XxlJobLogger.log("查询播主数={}",bloggers.size());
//        log.info("查询播主数={}",bloggers.size());
//        bloggers.forEach(e-> {
//            //查询已分析评论数比评论采集数小的，90天内视频列表
//            List<SimpleAweme> tmp = awemeRepository.findKeywordMessagesByUidAndRecentTimes(e.getUid(), condition);
//            tmp.forEach(simpleAweme -> {
//                List<AwemeComment> commentList=getCommentsFromS3(simpleAweme.getOAwemeId());
//                if(commentList==null||commentList.size()==0){
//                    return;
//                }
//                KeywordMessage keywordMessage=new KeywordMessage();
//                keywordMessage.setId(simpleAweme.getId());
//                keywordMessage.setOAwemeId(simpleAweme.getOAwemeId());
//                keywordMessage.setAwemeComments(commentList);
//                keywordMessage.setOCreateTime(simpleAweme.getOCreateTime());
//                keywordMessage.setRetryTimes(appConfig.getRetryKeywordTimes());
//                rst.add(keywordMessage);
//            });
//
//        });
        return rst;
    }
    public void send(int bloggerLogType, String who,TaskParam taskParam){
        long start=System.currentTimeMillis();
        //获取将发送的位置
        BloggerLog bloggerLog=getBloggerLog(bloggerLogType,who);
        //获取应该发送的消息
        //查询播主排行中的视频，
        List<SlimBloggerDto> bloggers = bloggerMainRepository.findSlimBloggersFollowerAndCount(appConfig.getSysCollectFollowerCount(),((int)bloggerLog.getCurrentBloggerId()-1)*taskParam.getRecords(),taskParam.getRecords());

        long condition = DateTimeUtils.second(LocalDateTime.now().minusDays(appConfig.getLatestDays()));
        List<KeywordMessage> rst=new ArrayList<>(taskParam.getRecords()*50);
        XxlJobLogger.log("查询播主数={}",bloggers.size());

        final Integer[] sendNums = {0};
        bloggers.forEach(e-> {
            //查询已分析评论数比评论采集数小的，90天内视频列表
            long queryStart=System.currentTimeMillis();
            List<SimpleAweme> tmp = awemeRepository.findKeywordMessagesByUidAndRecentTimes(e.getUid(), condition);
            tmp.forEach(simpleAweme -> {
                List<AwemeComment> commentList=getCommentsFromS3(simpleAweme.getOAwemeId());
                if(commentList==null||commentList.size()==0){
                    return;
                }
                KeywordMessage keywordMessage=new KeywordMessage();
                keywordMessage.setId(simpleAweme.getId());
                keywordMessage.setOAwemeId(simpleAweme.getOAwemeId());
                keywordMessage.setAwemeComments(commentList);
                keywordMessage.setOCreateTime(simpleAweme.getOCreateTime());
                keywordMessage.setRetryTimes(appConfig.getRetryKeywordTimes());
                kafkaTemplateKeyword.send(taskParam.getTopic(),keywordMessage);
                sendNums[0]++;
            });
            XxlJobLogger.log("发送数量={}，查询和发送时间={}s",tmp.size(),(System.currentTimeMillis()-queryStart)/1000);


        });
        //记录下一次发送的位置
        long nextPage=bloggerLog.getCurrentBloggerId()%taskParam.getPagesLimit()+1;
        if (sendNums[0] >0) {
            nextPage=taskParam.getStartPage();
            log.warn("页码={} ，无数据,下一次将从{}页开始",bloggerLog.getCurrentBloggerId(),nextPage);
            XxlJobLogger.log("页码={} ，无数据,下一次将从{}页开始",bloggerLog.getCurrentBloggerId(),nextPage);
        }


        //发送消息后，设置下一次发送位置
        bloggerLog.setCurrentBloggerId(nextPage);
        bloggerLogRepository.save(bloggerLog);
        log.info(" who={}, size={},page={}, pageSize={},  record from  {} to {},times={}秒",who,sendNums[0],bloggerLog.getCurrentBloggerId(),taskParam.getRecords(),(bloggerLog.getCurrentBloggerId()-1)*taskParam.getRecords(),bloggerLog.getCurrentBloggerId()*taskParam.getRecords(),(System.currentTimeMillis()-start)/1000);
        XxlJobLogger.log("  size={},page={}, pageSize={},  record from  {} to {},times={}秒",sendNums[0],bloggerLog.getCurrentBloggerId(),taskParam.getRecords(),(bloggerLog.getCurrentBloggerId()-1)*taskParam.getRecords(),bloggerLog.getCurrentBloggerId()*taskParam.getRecords(),(System.currentTimeMillis()-start)/1000);

    }

    @Override
    public void sendMsg(List<KeywordMessage> messages,String topic) {
        messages.forEach(msg->{
            msg.setRetryTimes(appConfig.getRetryKeywordTimes());
            msg.setRetryLimitTimes(appConfig.getRetryKeywordTimes());
            kafkaTemplateKeyword.send(topic,msg);
        });
        XxlJobLogger.log("投递主题={},投递消息数={}", topic,messages.size());
        log.info("投递主题={},投递消息数={}", topic,messages.size());
    }
    private List<AwemeComment> getCommentsFromS3(long awemeId){

        long start = System.currentTimeMillis();
        try{
            String comment = "";
            val aweme = awemeRepository.findByoAwemeId(awemeId);
            if(aweme.isPresent()){
                comment = awemeCommentRepository.get(awemeId,aweme.get().getDatecode());
            }
            if((System.currentTimeMillis()-start)/1000>appConfig.getDebugTimeLimit())
                log.info(" get oss data awemeId:{} use time:{}s",awemeId,(System.currentTimeMillis()-start)/1000);
            final ObjectMapper mapper = new ObjectMapper();
            List<AwemeComment> awemeCommentList = mapper.readValue(comment, new TypeReference<>() {});
            return awemeCommentList;

        }catch (Exception e){
            log.error("getCommentsFromOss awemeId:{} use time:{}s,{}",awemeId,(System.currentTimeMillis()-start)/1000,e.getMessage());
        }

        return Collections.emptyList();

    }
}
